package Source

import org.apache.commons.math3.random.{GaussianRandomGenerator, JDKRandomGenerator}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import java.io.PrintWriter
import java.util.{Properties, Random}

/**
 * 龚英杰：
 * 模拟生成数据
 * 将模拟的数据生成到文件中
 * 将模拟的数据生成到kafka中
 *
 * 区域id、道路id、卡口id、摄像头id、拍摄时间、车辆信息、车辆速度
 */

object MakeData {
  def main(args: Array[String]): Unit = {

    val pw = new PrintWriter("src/main/scala/data/trafficdata")
    val sqlw = new PrintWriter("src/main/scala/data/tra.sql")

    val prop = new Properties()
    prop.put("bootstrap.servers","192.168.23.62:9092,192.168.23.66:9092,192.168.23.68:9092")
    prop.put("key.serializer",classOf[StringSerializer].getName)
    prop.put("value.serializer",classOf[StringSerializer].getName)

    val producer = new KafkaProducer[String, String](prop)

    val locations = Array[String]("京","津","冀","京","鲁","京")
    val area = Array[String]("01","02","03")
    val road = Array[String]("10","11","12","13")
    val mon = Array[String]("0000","0001","0002","0003")

    val random = new Random()
    val generator = new GaussianRandomGenerator(new JDKRandomGenerator())

//    模拟车辆
    for(i <- 1 to 200){
      val car = locations(random.nextInt(locations.length)) + (65 + random.nextInt(26)).toChar+random.nextInt(99999).formatted("%05d")
      val throuldMonitorCount = (generator.nextNormalizedDouble() * 100).abs.toInt
      //    模拟通过的卡口，最多100
      for (j <- 1 until throuldMonitorCount) {
        //      区域
        val areaId = area(random.nextInt(3).toInt)
        //      道路
        val roadId = road(random.nextInt(4))
        //      卡口
        val monitorId = mon(random.nextInt(4))
        //      摄像头id
        val cameraId = random.nextInt(99999).formatted("%05d")
        //      拍摄时间
        val actionId = DateUtils.getCurrentDate() + " " + DateUtils.getRandomHour() + ":" + DateUtils.getRandomMorS() + ":" + DateUtils.getRandomMorS()
        //      速度
        val speed = random.nextInt(140)

        val info = s"${areaId}\t${roadId}\t${monitorId}\t${cameraId}\t${actionId}\t${car}\t${speed}"
        println(info)
        val sqlinfo = s"insert into hhd.tra value('${areaId}','${roadId}','${monitorId}','${cameraId}','${actionId}','${car}','${speed}');"
        println(sqlinfo)
        sqlw.println(sqlinfo)

        pw.println(info)

        producer.send(new ProducerRecord[String,String]("test",info))
      }
    }

    pw.close()
    sqlw.close()
    producer.close()

  }
}
